package com.example.queuetest;


import com.example.util.ConnectionUtils;
import com.rabbitmq.client.*;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

public class QueueTest {

    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
        Connection connection1 = ConnectionUtils.openConnection("c1");
        Connection connection2 = ConnectionUtils.openConnection("c2");

        Channel channel1 = connection1.createChannel();
        Channel channel2 = connection2.createChannel();

        String queueName = "test-queue";
        boolean durable = false;
        boolean exclusive = false;
        boolean autoDelete = true;
        AMQP.Queue.DeclareOk ok1 = channel1.queueDeclare(queueName, durable, exclusive, autoDelete, Map.of(
                "x-max-length", 2,
                //"x-overflow", "reject-publish",
                //"x-dead-letter-exchange","dead-exchange" ,
                "x-single-active-consumer", true));

        AMQP.BasicProperties properties = new AMQP.BasicProperties("text/plain",
                null,
                null,
                2,
                0, null, null, null,
                "001", null, null, null,
                null, null);

        channel1.confirmSelect();

        //ConcurrentMap<Long, String> map = new ConcurrentHashMap<>();
        //long nextPublishSeqNo = channel1.getNextPublishSeqNo();
        //System.out.println("nextPublishSeqNo: " + nextPublishSeqNo);
        //
        //map.put(nextPublishSeqNo, "This is the message for test");


        //channel1.addConfirmListener(new ConfirmListener() {
        //    @Override
        //    public void handleAck(long deliveryTag, boolean multiple) throws IOException {
        //
        //    }
        //
        //    @Override
        //    public void handleNack(long deliveryTag, boolean multiple) throws IOException {
        //        System.out.println("deliveryTag:" + deliveryTag);
        //        channel1.basicPublish("", "test-queue", properties, map.get(deliveryTag).getBytes(StandardCharsets.UTF_8));
        //    }
        //});

        channel1.addReturnListener(new ReturnListener() {
            @Override
            public void handleReturn(int replyCode, String replyText, String exchange, String routingKey, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("return listener handle return: " + new String(body, StandardCharsets.UTF_8));
            }
        });

        for (int i = 0; i < 5; i++) {
            channel2.basicPublish("", "test-queue", true, properties, String.format("msg%d", i).getBytes(StandardCharsets.UTF_8));
        }

        //channel1.basicPublish("", "test-queue", properties, "msg2".getBytes(StandardCharsets.UTF_8));


        //channel1.basicPublish("", "test-queue", MessageProperties.PERSISTENT_TEXT_PLAIN, "some msg02".getBytes(StandardCharsets.UTF_8));
        //channel1.basicPublish("", "test-queue", MessageProperties.PERSISTENT_TEXT_PLAIN, "some msg03".getBytes(StandardCharsets.UTF_8));

        //channel1.basicQos(1);
        //
        long start = System.currentTimeMillis();
        //channel1.basicConsume("test-queue",false, (consumerTag, message) -> { // 会抛异常：cannot obtain exclusive access to locked queue 'test-queue' in vhost '/'. It could be originally declared on another connection or the exclusive property value does not match that of the original declaration.
        //    System.out.println("消费者1 receive msg: " + new String(message.getBody(), StandardCharsets.UTF_8));
        //    System.out.println("耗时：" + (System.currentTimeMillis() - start) + "毫秒");
        //    channel1.basicAck(message.getEnvelope().getDeliveryTag(), false);
        //}, consumerTag -> {
        //
        //});
        //
        channel1.basicConsume("test-queue",false, (consumerTag, message) -> { // 会抛异常：cannot obtain exclusive access to locked queue 'test-queue' in vhost '/'. It could be originally declared on another connection or the exclusive property value does not match that of the original declaration.
            System.out.println("消费者2 receive msg: " + new String(message.getBody(), StandardCharsets.UTF_8));
            System.out.println("耗时：" + (System.currentTimeMillis() - start) + "毫秒");
            channel1.basicAck(message.getEnvelope().getDeliveryTag(), false);
        }, consumerTag -> {

        });


        //String tempQueue = channel1.queueDeclare("test-auto-del", false, false, true, null).getQueue();
        //channel1.queueBind(tempQueue, "", tempQueue);
        ////
        //channel1.basicConsume(tempQueue, (consumerTag, message) -> { // 会抛异常：cannot obtain exclusive access to locked queue 'test-queue' in vhost '/'. It could be originally declared on another connection or the exclusive property value does not match that of the original declaration.
        //    System.out.println("temp Queue receive msg: " + new String(message.getBody(), StandardCharsets.UTF_8));
        //}, consumerTag -> {
        //
        //});

        TimeUnit.SECONDS.sleep(5000);

        connection1.close();
        connection2.close();
    }
}
